package com.flink.boradcast;

import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;



/**
 * 描述:
 * 广播流 示例
 *  TODO 广播状态生效执行顺序 1.13之前和 1.14以后有重大改变 1.13之前当前keyby分区执行完毕生效，1.14以后立即生效！
 * @author yanzhengwu
 * @create 2022-09-04 15:23
 */
public class BroadcastDemo {

    public static class Action {
        public String userId;
        public String action;

        @Override
        public String toString() {
            return "Action{" +
                    "userId='" + userId + '\'' +
                    ", action='" + action + '\'' +
                    '}';
        }

        public Action() {
        }

        public Action(String userId, String action) {
            this.userId = userId;
            this.action = action;
        }
    }

    public static class Pattern {
        @Override
        public String toString() {
            return "Pattern{" +
                    "action1='" + action1 + '\'' +
                    ", action2='" + action2 + '\'' +
                    '}';
        }

        public String action1;
        public String action2;

        public Pattern() {
        }

        public Pattern(String action1, String action2) {
            this.action1 = action1;
            this.action2 = action2;
        }
    }

    public static void main(String[] args) throws Exception {
        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        env.setParallelism(1);

        DataStreamSource<Action> actionStream = env.fromElements(
                new Action("Alice", "login"),
                new Action("Alice", "pay"),
                new Action("Bob", "login"),
                new Action("Bob", "order")
                //new Action("Bob", "prod")

        );
        //用户规则，是先登录再支付 还是先登录再购买
        DataStreamSource<Pattern> patternBroadStream = env.fromElements(
                new Pattern("login", "pay"),
                new Pattern("login", "order")
        );


        //定义广播流描述器
        MapStateDescriptor<Void, Pattern> mapStateDescriptor = new MapStateDescriptor<Void, Pattern>("pattern", Types.VOID, Types.POJO(Pattern.class));
        //添加广播状态
        BroadcastStream<Pattern> broadcastStream = patternBroadStream.broadcast(mapStateDescriptor);

        SingleOutputStreamOperator<String> singleOutputStreamOperator = actionStream
                .keyBy(data -> data.userId)
                .connect(broadcastStream)
                .process(new PatternDetector());

        singleOutputStreamOperator.print();


        env.execute();
    }


    public static class PatternDetector extends KeyedBroadcastProcessFunction<String, Action, Pattern, String> {
        //保存上次用户的行为
        ValueState<String> prevActionState;

        @Override
        public void open(Configuration parameters) throws Exception {
            prevActionState = getRuntimeContext().getState(new ValueStateDescriptor<String>("last-action", String.class));
        }

        //只能获取广播状态不能更改
        @Override
        public void processElement(Action value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
            //System.out.println("处理数据：" + value);

            ReadOnlyBroadcastState<Void, Pattern> patternState = ctx.getBroadcastState(new MapStateDescriptor<Void, Pattern>("pattern", Types.VOID, Types.POJO(Pattern.class)));
            //规则状态
            Pattern pattern = patternState.get(null);
            //System.out.println("规则：" + pattern);
            //上次用户状态
            String prevAction = prevActionState.value();
            //System.out.println("获取到的上次用户状态：" + prevAction);
            if (pattern != null && prevAction != null) {
                if (pattern.action1.equals(prevAction) && pattern.action2.equals(value.action)) {
                    out.collect("处理结果->  " + ctx.getCurrentKey() + " : " + pattern.toString());
                }
            }
            prevActionState.update(value.action);
            //System.out.println("更新后的当前用户状态：" + prevActionState.value());

            //System.out.println("------------------------");

        }

        //可以更新广播状态和获取广播状态
        @Override
        public void processBroadcastElement(Pattern value, Context ctx, Collector<String> out) throws Exception {
            BroadcastState<Void, Pattern> patternState = ctx.getBroadcastState(new MapStateDescriptor<Void, Pattern>("pattern", Types.VOID, Types.POJO(Pattern.class)));
            //更新状态
            patternState.put(null, value);
            //Thread.sleep(10000);
            //System.out.println("更新广播状态：" + value);

        }


    }
}
